/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package homomorphism.NewAPI;

import homomorphism.HomomorphismBase;
import java.io.IOException;
import java.lang.reflect.*;
import java.util.ArrayList;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import utility.LongPair;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.mapreduce.Mapper;

/**
 *
 * @author takeshi
 */
public abstract class ListIndexingMapper<InVal, OtVal> extends Mapper<LongWritable, InVal, LongPair, InVal> {

    //  private static final Log LOG = LogFactory.getLog(ListIndexingMapper.class);
    public static final Log LOG = LogFactory.getLog(ListIndexingMapper.class);
    private static final String userClassPro = HomomorphismBase.UserClassNameProperty;
//    private static final String InValClassPro = HomomorphismBase.InKeyClassProperty;//"mapred.input.value.class";
    //   private static final String OtValClassPro = HomomorphismBase.IndexKeyClassProperty;//"mapred.output.value.class";
    public static final int Mega = HomomorphismBase.Mega;
    private Class FilterType;
    private Class AggType;
    private String UserClassName;
    private long lastpid, orIndex;
    private ArrayList<InVal> localCache;
    LongPair outKeyPair = new LongPair(-1, -1);

    static public int makePid(long id) {
        int pid = (int) id >> 10;
        return pid;
    }

    @Override
    public void setup(Context context) throws IOException, InterruptedException {

        lastpid = -1;
        UserClassName = context.getConfiguration().get(userClassPro);

        localCache = new ArrayList<InVal>();
        try {
            FilterType = context.getMapperClass().getTypeParameters()[0].getClass();

        } catch (ClassNotFoundException ex) {
            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
        }
        AggType = context.getMapOutputValueClass(); //==IndexKeyClassProperty  assertThat()...

    }

    @Override
    public void cleanup(Context context) throws IOException, InterruptedException {
        try {
            int count = 0;
            for (InVal e : compressCache(context)) {

                outKeyPair.Set(lastpid, orIndex + count);
                count++;
                context.write(outKeyPair, e);
            }
        } catch (Exception ex) {
            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
        }
        localCache.clear();
    }

    public OtVal filter(InVal in) throws ClassNotFoundException, NoSuchMethodException, InstantiationException, IllegalAccessException, IllegalArgumentException, InvocationTargetException {

        OtVal rst = null;
        Class cls = Class.forName(UserClassName);
        Class partypes[] = new Class[1];

        partypes[0] = FilterType;
        try {
            Method _filter = cls.getMethod("filter", partypes);
            if (_filter != null) {
                Object arglist[] = new Object[1];
                arglist[0] = in;
                Constructor ct = cls.getConstructor(); //partypes
                Object methobj = ct.newInstance();
                rst = (OtVal) _filter.invoke(methobj, arglist);
            }
        } catch (NoSuchMethodException e) {
            //no such a method then do nothing
            rst = (OtVal) in;
        }
        return rst;
    }

    public OtVal fold(ArrayList<InVal> values)
            throws ClassNotFoundException, IOException,
            NoSuchMethodException, InstantiationException,
            IllegalAccessException, IllegalArgumentException,
            InvocationTargetException {

        if (UserClassName == null || UserClassName.isEmpty()) {
            throw new IOException("1null  fold:UserClassName\n");
        }
        // try {
        Class UserApp = Class.forName(UserClassName);
        Class partypes[] = new Class[1];
        partypes[0] = ArrayList.class;
        Method _fold = UserApp.getMethod("fold", partypes);

        Object arglist[] = new Object[1];
        arglist[0] = values;
        Constructor ct = UserApp.getConstructor(); //partypes
        Object methobj = ct.newInstance();
        OtVal rst = (OtVal) _fold.invoke(methobj, arglist);
        return rst;

//        } catch (ClassNotFoundException ee) {
//            LOG.error(ee.getMessage());
//        }
        // return null;

    }

    public ArrayList<InVal> unfolding(OtVal value) throws ClassNotFoundException, NoSuchMethodException,
            InstantiationException, IllegalAccessException, IllegalArgumentException, InvocationTargetException {
        ClassLoader myloader = ListIndexingMapper.class.getClassLoader();
        Class UserApp;
        if (UserClassName == null || UserClassName.isEmpty()) {
            return null;
        }
//        try {
        UserApp = myloader.loadClass(UserClassName);
        Class partypes[] = new Class[1];
        partypes[0] = value.getClass(); // UnFoldingType;
        //for debug/test
        //   assert partypes[0].getSimpleName().equalsIgnoreCase(AggType.getSimpleName()) : "Something wrong, type not matched! ";
        Method _unfold = UserApp.getMethod("unfold", partypes);
        Object arglist[] = new Object[1];
        arglist[0] = value;
        Constructor ct = UserApp.getConstructor();//partypes
        Object methobj = ct.newInstance();
        ArrayList<InVal> rst = (ArrayList<InVal>) _unfold.invoke(methobj, arglist);
        return rst;
//        } catch (ClassNotFoundException ee) {
//            LOG.error(ee.getMessage());
//        }
        //  return null;
    }

    @Override
    public void map(LongWritable key, InVal value, Context context) throws IOException, InterruptedException {

        if (value == null) {
            return;
        }
        long OrIndex = key.get();
        long thispid = makePid(OrIndex); //(OrIndex / Mega);
        if (lastpid < thispid) {
            //assert the size of localCache is 65536 ??
            orIndex = OrIndex;
            if (lastpid >= 0) {
                try {
                    int count = 0;
                    for (InVal e : compressCache(context)) {
                        outKeyPair.Set(thispid, OrIndex + count);
                        count++;
                        context.write(outKeyPair, e);
                    }
                } catch (ClassNotFoundException ex) {
                    Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
                } catch (NoSuchMethodException ex) {
                    Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
                } catch (InstantiationException ex) {
                    Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
                } catch (IllegalAccessException ex) {
                    Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
                } catch (IllegalArgumentException ex) {
                    Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
                } catch (InvocationTargetException ex) {
                    Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
                }

            }


            // need to know when the last time pid cheanged, what is the OrIndex.

        }
//        if (OrIndex < pid) {
//            throw new IOException("FirstMapper: OrIndex < pid\n");
//        }
        lastpid = thispid;
        localCache.add(value);
//          Constructor ct;
//          Object v = null;
//        try {
//            ct = value.getClass().getConstructor(); //partypes
//             v= ct.newInstance();
//        } catch (InstantiationException ex) {
//            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
//        } catch (IllegalAccessException ex) {
//            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
//        } catch (IllegalArgumentException ex) {
//            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
//        } catch (InvocationTargetException ex) {
//            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
//        } catch (NoSuchMethodException ex) {
//            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
//        } catch (SecurityException ex) {
//            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
//        }
//          v = (Object)value;
//        localCache.add((InVal) v);

//        if (IsFunctionDefined("filter")) {
//            try {
//                //if there is a filter function
//                context.write(outKeyPair, filter(value));
//
//            } catch (Exception ex) {
//                Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
//            }
//        } else {
//            context.write(outKeyPair, (OtVal) value);
//        }

        //           catch (NoSuchMethodException ex) {
//            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
//        } catch (InstantiationException ex) {
//            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
//        } catch (IllegalAccessException ex) {
//            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
//        } catch (IllegalArgumentException ex) {
//            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
//        } catch (InvocationTargetException ex) {
//            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
//        }
    }

    private boolean IsFunctionDefined(final String name) {
        Class cls = null;
        try {
            cls = Class.forName(UserClassName);
        } catch (ClassNotFoundException ex) {
            Logger.getLogger(ListIndexingMapper.class.getName()).log(Level.SEVERE, null, ex);
        }
        Method[] funcs = cls.getDeclaredMethods();
        for (Method f : funcs) {
            if (f.getName().equals(name)) {
                return true;
            }
        }
        return false;
    }

    private ArrayList<InVal> compressCache(Context context) throws IOException, InterruptedException,
            ClassNotFoundException, NoSuchMethodException, InstantiationException,
            IllegalAccessException, IllegalArgumentException, InvocationTargetException {

        if (localCache.isEmpty()) {
            return null;
        }
        ArrayList<InVal> rst = new ArrayList<InVal>();

        //   if (IsFunctionDefined("filter")) {
        //context.write(outKeyPair, fold(localCache));
        OtVal finalV = fold(localCache);
        localCache.clear();
        for (InVal value : unfolding(finalV)) {
            rst.add(value);
            //context.write(outKeyPair, (OtVal) value);
        }
        // context.progress();
        return rst;
        //   }
    }
}
